import { Writable, WritableOptions } from 'stream';
import fs from 'fs';
import { open } from 'fs/promises';

const DEFAULT_HIGH_WATER_MARK = 16 * 1024 * 1024; // 16MB
const MAX_BUFFER_GC_SIZE = 10 * 1024 * 1024; // 10MB

interface FileDescriptor {
  fd: fs.promises.FileHandle;
  path: string;
  opened: boolean;
  pendingWrites: number;
}

class StreamWriter extends Writable {
  // 内存模式缓冲区
  private bufferList: Buffer[] = [];
  private bufferSize = 0;
  
  // 文件模式资源管理
  private fileDescriptor: FileDescriptor | null = null;
  private writeQueue: Promise<void>[] = [];
  private pendingOperations = 0;
  
  // 流状态控制
  private explicitCloseRequested = false;
  private ended = false;
  private streamClosed = false;
  private streamDestroyed = false;
  private lastError: Error | null = null;
  
  // 性能计数器
  private bytesWritten = 0;
  private writesCount = 0;

  constructor(target?: string, opts: WritableOptions = {}) {
    super({
      autoDestroy: true,
      emitClose: true,
      highWaterMark: DEFAULT_HIGH_WATER_MARK,
      ...opts,
      decodeStrings: true
    });
    
    if (target) {
      this.fileDescriptor = {
        fd: null as any, // 初始化占位
        path: target,
        opened: false,
        pendingWrites: 0
      };
    }
  }

  /**
   * 预分配文件句柄 (可选)
   * 可在实际写入前提前打开文件，提升首条写入性能
   */
  async preopen(): Promise<void> {
    if (!this.fileDescriptor || this.explicitCloseRequested) {
      return;
    }
    
    if (!this.fileDescriptor.opened) {
      try {
        this.fileDescriptor.fd = await open(this.fileDescriptor.path, 'w');
        this.fileDescriptor.opened = true;
      } catch (err) {
        this.destroy(err as Error);
        throw err;
      }
    }
  }

  /**
   * 写入数据实现 (Node.js 流的内部方法)
   */
  _write(
    chunk: Buffer | string, 
    encoding: BufferEncoding,
    callback: (error?: Error | null) => void
  ): void {
    if (this.streamDestroyed) {
      callback(new Error('Stream has been destroyed'));
      return;
    }
    
    const buffer = this._convertToBuffer(chunk, encoding);
    
    if (this.fileDescriptor) {
      this._fileWrite(buffer, callback);
    } else {
      this._inMemoryWrite(buffer, callback);
    }
  }

  /**
   * 批量写入优化 (Node.js 流的内部方法)
   */
  _writev(
    chunks: { chunk: Buffer | string; encoding: BufferEncoding }[], 
    callback: (error?: Error | null) => void
  ): void {
    if (this.streamDestroyed) {
      callback(new Error('Stream has been destroyed'));
      return;
    }
    
    const buffers = chunks.map(chunk => 
      this._convertToBuffer(chunk.chunk, chunk.encoding)
    );
    
    // 合并小缓冲区减少系统调用
    const totalSize = buffers.reduce((sum, buf) => sum + buf.length, 0);
    
    if (this.fileDescriptor) {
      // 修复类型错误：直接使用缓冲区数组
      if (buffers.length > 1 && totalSize < 4096) {
        // 多个小缓冲区合并为单个写入
        const combined = Buffer.concat(buffers);
        this._fileWrite(combined, callback);
      } else {
        this._fileWritev(buffers, callback);
      }
    } else {
      this.bufferList.push(...buffers);
      this.bufferSize += totalSize;
      this.bytesWritten += totalSize;
      this.writesCount += chunks.length;
      process.nextTick(callback);
    }
  }

  /**
   * 写入一行文本（自动添加换行符）
   * @param line 要写入的文本内容
   */
  writeLine(line: string): boolean {
    return this.write(`${line}\n`);
  }

  /**
   * 清空内存缓冲区（仅内存模式）
   */
  resetBuffer(): void {
    if (this.fileDescriptor) {
      throw new Error('Cannot reset buffer in file mode');
    }
    
    this.bufferList = [];
    this.bufferSize = 0;
    this.bytesWritten = 0;
  }

  /**
   * 安全关闭资源
   */
  async close(): Promise<void> {
    if (this.streamClosed || this.streamDestroyed) {
      return;
    }
    
    this.explicitCloseRequested = true;
    
    // 等待所有操作完成
    await this._drainQueue();
    
    if (this.fileDescriptor?.opened) {
      try {
        await this.fileDescriptor.fd.close();
        this.fileDescriptor.opened = false;
      } catch (err) {
        this.destroy(err as Error);
        throw err;
      }
    }
    
    this.streamClosed = true;
    this.emit('closed');
  }

  /**
   * 获取内存缓冲区内容
   */
  getBuffer(encoding?: BufferEncoding): string | Buffer {
    if (this.fileDescriptor) {
      throw new Error('Cannot get buffer in file mode');
    }
    
    const data = Buffer.concat(this.bufferList);
    return encoding ? data.toString(encoding) : data;
  }

  //==================== 私有方法 ====================//

  private _convertToBuffer(chunk: Buffer | string, encoding: BufferEncoding): Buffer {
    return Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk, encoding);
  }

  private async _fileWrite(buffer: Buffer, callback: (error?: Error | null) => void) {
    const writeOp = this._executeFileWrite(buffer)
      .then(() => callback())
      .catch(err => callback(err));
    
    this.writeQueue.push(writeOp);
    this._manageQueue();
  }

  private async _fileWritev(buffers: Buffer[], callback: (error?: Error | null) => void) {
    const writeOp = this._executeFileWritev(buffers)
      .then(() => callback())
      .catch(err => callback(err));
    
    this.writeQueue.push(writeOp);
    this._manageQueue();
  }

  private async _executeFileWrite(buffer: Buffer): Promise<void> {
    if (this.streamDestroyed) {
      throw new Error('Stream destroyed during write operation');
    }

    // 添加健壮性检查
    if (!this.fileDescriptor) {
      throw new Error('File descriptor does not exist');
    }

    if (!this.fileDescriptor.opened) {
      try {
        this.fileDescriptor.fd = await open(this.fileDescriptor.path, 'w');
        this.fileDescriptor.opened = true;
      } catch (err) {
        this.destroy(err as Error);
        throw err;
      }
    }

    let retries = 0;
    let lastError: Error | null = null;
    
    const writeWithRetry = async (): Promise<void> => {
      // 每次重试前检查状态
      if (this.streamDestroyed || !this.fileDescriptor?.opened) {
        throw new Error('Stream closed during retry operation');
      }
      
      try {
        await this.fileDescriptor.fd.write(buffer);
        this.bytesWritten += buffer.length;
        this.writesCount++;
        this.fileDescriptor.pendingWrites--;
      } catch (err) {
        // 特定错误重试（如资源忙）
        if (this._isRetryableError(err) && retries < 3) {
          retries++;
          await new Promise(resolve => setTimeout(resolve, 10 * retries));
          return writeWithRetry();
        }
        lastError = err as Error;
      }
    };
    
    this.fileDescriptor.pendingWrites++;
    await writeWithRetry();
    
    if (lastError) {
      throw lastError;
    }
  }

  private async _executeFileWritev(buffers: Buffer[]): Promise<void> {
    if (this.streamDestroyed) {
      throw new Error('Stream destroyed during writev operation');
    }

    // 添加健壮性检查
    if (!this.fileDescriptor) {
      throw new Error('File descriptor does not exist');
    }

    if (!this.fileDescriptor.opened) {
      try {
        this.fileDescriptor.fd = await open(this.fileDescriptor.path, 'w');
        this.fileDescriptor.opened = true;
      } catch (err) {
        this.destroy(err as Error);
        throw err;
      }
    }

    try {
      // 修复类型错误：直接使用缓冲区数组
      await this.fileDescriptor.fd.writev(buffers);
      const totalBytes = buffers.reduce((sum, buf) => sum + buf.length, 0);
      this.bytesWritten += totalBytes;
      this.writesCount++;
    } catch (err) {
      this.destroy(err as Error);
      throw err;
    }
  }

  private _inMemoryWrite(buffer: Buffer, callback: (error?: Error | null) => void) {
    this.bufferList.push(buffer);
    this.bufferSize += buffer.length;
    this.bytesWritten += buffer.length;
    this.writesCount++;
    
    // 自动内存管理
    if (this.bufferSize > MAX_BUFFER_GC_SIZE && this.bufferList.length > 4) {
      // 合并小缓冲区减少内存碎片
      const newBuffer = Buffer.concat(this.bufferList);
      this.bufferList = [newBuffer];
      this.bufferSize = newBuffer.length;
    }
    
    process.nextTick(callback);
  }

  private async _drainQueue(): Promise<void> {
    while (this.writeQueue.length > 0 || this.pendingOperations > 0) {
      // 等待队列中的写入操作完成
      if (this.writeQueue.length > 0) {
        // 修复：避免等待非Promise值
        const queueCopy = [...this.writeQueue];
        this.writeQueue = [];
        await Promise.all(queueCopy);
      }
      
      // 等待文件描述符上的操作完成
      
      if (this.fileDescriptor?.pendingWrites && this.fileDescriptor?.pendingWrites > 0) {
        await new Promise(resolve => setImmediate(resolve));
      }
    }
  }

  private async _manageQueue(): Promise<void> {
    // 控制最大并发写入操作
    if (this.pendingOperations < 5 && this.writeQueue.length > 0) {
      this.pendingOperations++;
      const op = this.writeQueue.shift();
      
      if (op) { // 避免等待undefined
        try {
          await op;
        } finally {
          this.pendingOperations--;
          this._manageQueue(); // 处理下一个写入操作
        }
      }
    }
  }

  private _isRetryableError(err: any): boolean {
    return err?.code && ['EAGAIN', 'EBUSY', 'EINTR'].includes(err.code);
  }

  //==================== 生命周期管理 ====================//

  _final(callback: (error?: Error | null) => void): void {
    this.ended = true;
    
    const finalize = async () => {
      try {
        await this._drainQueue();
        if (!this.explicitCloseRequested) {
          await this.close();
        }
        callback();
      } catch (err) {
        callback(err as Error);
      }
    };
    
    finalize().catch(callback);
  }

  _destroy(error: Error | null, callback: (error?: Error | null) => void): void {
    this.streamDestroyed = true;
    this.lastError = error;
    
    const destroyInternal = async () => {
      try {
        if (this.fileDescriptor?.opened) {
          try {
            await this.fileDescriptor.fd.close();
          } catch (closeErr) {
            // 优先记录主要错误
            error = error || (closeErr as Error);
          }
        }
        
        if (error) this.emit('error', error);
        callback(error || null);
      } finally {
        // 释放内存资源
        this.bufferList = [];
        this.bufferSize = 0;
      }
    };
    
    // 确保异步销毁过程有返回值
    destroyInternal().catch(err => callback(err));
  }
  
  getBytesWritten(): number {
    return this.bytesWritten;
  }
  
  getWriteCount(): number {
    return this.writesCount;
  }
}

export default StreamWriter;

